package com.example.hbasedemo;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.ByteArrayHashKey;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.MurmurHash;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hbase.thirdparty.io.netty.util.concurrent.DefaultThreadFactory;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.io.IOException;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
import java.util.stream.Stream;

@Slf4j
@SpringBootApplication
public class HbaseDemoApplication {

    public static final String HBASE_QUORUM = "hbase.zookeeper.quorum";
    public static final String HBASE_CLIENT_PORT = "hbase.zookeeper.property.clientPort";
    public static final String HBASE_CLIENT_RETRIES_NUMBER = "hbase.client.retries.number";
    public static final String HBASE_RPC_TIMEOUT = "hbase.rpc.timeout";
    public static final String HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD = "hbase.client.scanner.timeout.period";
    public static final String HBASE_CLIENT_OPERATION_TIMEOUT = "hbase.client.operation.timeout";

    private static byte[][] getSplitKeys() {
        String[] keys = new String[]{"10|", "20|", "30|", "40|", "50|","60|", "70|", "80|", "90|"};
        byte[][] splitKeys = new byte[keys.length][];
        TreeSet<byte[]> rows = new TreeSet<>(Bytes.BYTES_COMPARATOR);//升序排序
        for (String key : keys) {
            rows.add(Bytes.toBytes(key));
        }
        Iterator<byte[]> rowKeyIter = rows.iterator();
        int i = 0;
        while (rowKeyIter.hasNext()) {
            byte[] tempRow = rowKeyIter.next();
            rowKeyIter.remove();
            splitKeys[i] = tempRow;
            i++;
        }
        return splitKeys;
    }

    private static String getDevicePrefix2(String deviceId) {
        byte[] bytes = deviceId.getBytes();
        int hashCode = MurmurHash.getInstance().hash(new ByteArrayHashKey(bytes, 0, bytes.length), -1);
        if(hashCode < 0) hashCode = 0 - hashCode;
        return String.format("%02d%20s", hashCode % 100, deviceId).replace(' ', '0');
    }

    public static Connection createConnection(String quorum, int port) throws IOException {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set(HBASE_QUORUM, quorum);
        configuration.set(HBASE_CLIENT_PORT, String.valueOf(port));
        return ConnectionFactory.createConnection(configuration);
    }

    public static void createTable(Connection connection, String tableName, List<String> families, BiConsumer<String,ColumnFamilyDescriptorBuilder> init) throws IOException {
        try(final Admin admin = connection.getAdmin()) {
            final byte[][] splitKeys =  getSplitKeys();
            TableName table = TableName.valueOf(tableName);
            if(admin.tableExists(table)) {
                TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(table);
                Collection<ColumnFamilyDescriptor> columnFamilyDescriptors = new ArrayList<>(families.size());
                families.forEach(k -> {
                    ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(k.getBytes());
                    // 启用压缩
                    columnFamilyDescriptorBuilder.setCompressionType(Compression.Algorithm.SNAPPY);
                    // 设置最大版本数,为1表示每次更新直接覆盖旧版本
                    columnFamilyDescriptorBuilder.setMaxVersions(1);
                    // 设置数据的备份数
                    columnFamilyDescriptorBuilder.setDFSReplication((short) 1);
                    if(init != null) {
                        init.accept(k, columnFamilyDescriptorBuilder);
                    }
                });
                tableDescriptorBuilder.setColumnFamilies(columnFamilyDescriptors);
                admin.createTable(tableDescriptorBuilder.build(),splitKeys);
            }

            // BufferedMutator用于批量写数据insert or delete
            BufferedMutatorParams params = new BufferedMutatorParams(table);
                    /*
                     if(WriteBufferPeriodicFlushTimeoutMs > 0) {
                        new Timer(() -> {
                            if(firstRecorderTimestamp + WriteBufferPeriodicFlushTimeoutMs <= now) {
                                flush();
                            }
                        }, WriteBufferPeriodicFlushTimerTickMs, WriteBufferPeriodicFlushTimerTickMs)
                     }
                     if(currentBufferSize = 0) {
                        firstRecorderTimestamp = now;
                     }
                     if(currentBufferSize > bufferSize) {
                        flush();
                     }
                     */
            params.setWriteBufferPeriodicFlushTimeoutMs(1000);
            params.setWriteBufferPeriodicFlushTimerTickMs(5000);
            params.writeBufferSize(1024 * 512);
//            params.listener(new BufferedMutator.ExceptionListener() {
//                @Override
//                public void onException(RetriesExhaustedWithDetailsException e, BufferedMutator bufferedMutator) throws RetriesExhaustedWithDetailsException {
//                    // 异常处理
//                }
//            });
//            connection.getBufferedMutator(params);
        }
    }

    public static void readRow() throws IOException {
        //        ImmutableBytesWritable

        try(Connection client = createConnection("10.94.44.91,10.94.44.92,10.94.44.93", 2181)) {
            Table table = client.getTable(TableName.valueOf("trackerGSensor"));
            Scan scan = new Scan();
            String deviceId = "869374040260282";
            String rowKey = getDevicePrefix2(deviceId);
            System.out.println(rowKey);
            scan.setFilter(new PrefixFilter(rowKey.getBytes()));
            try(ResultScanner scanner = table.getScanner(scan)) {
                // Reading values from scan result
                for (Result result = scanner.next(); result != null; result = scanner.next()) {
                    System.out.println("Found row : " + result);
                }
            }
        }
    }




    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {


    }

}
